package com.qf.demo1.conf;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.*;

@Configuration
public class MqttConfig {

    /*
        入站的通道，设备发送过来的消息
     */
    @Bean
    public MessageChannel inChannel(){
        return new DirectChannel();
    }
    /*
        出站的通道，程序向设备发送的消息
     */
    @Bean
    public MessageChannel outChannel(){
        return new DirectChannel();
    }

    /*
        入站的消息处理器
     */
    @Bean
    @ServiceActivator(inputChannel = "inChannel")
    public MessageHandler inMessageHandler(){
        return msg->{
            //收到设备上报消息后，要用消息中间件实现削峰填谷
            //将收到的入站消息转到service层进行业务处理
            MessageHeaders headers = msg.getHeaders();
            System.out.println(headers);
            Object payload = msg.getPayload();
            System.out.println(payload);

            System.out.println("--------------------");
        };
    }

    /*
      出站的消息处理器
    */
    @Bean
    @ServiceActivator(inputChannel = "outChannel")
    public MessageHandler outMessageHandler(MqttPahoClientFactory factory){
        MqttPahoMessageHandler handler = new MqttPahoMessageHandler("demo1_out_mqtt",factory);

        //设置默认主题
        handler.setDefaultTopic("java2303/default/#");
        //设置默认qos等级
        handler.setDefaultQos(1);

        //支持异步发送（提速）
        handler.setAsync(true);

        return handler;
    }

    @Bean
    public MessageProducer messageProducer(MqttPahoClientFactory factory,MessageChannel inChannel){
        //当前作为mqtt客户端的唯一性标识
        String clientId = "demo1_in_mqtt";
        //当前客户端订阅订单主题
        String[] topics = {"java2303/#","java2302/#"};
        //连接mqtt服务器的连接工厂
        MqttPahoClientFactory connFactory = factory;

        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(clientId,connFactory,topics);

        //设置收到的消息往哪里通道去
        adapter.setOutputChannel(inChannel);
        //消息服务质量等级
        adapter.setQos(1);
        //设置消息转换器
        adapter.setConverter(new DefaultPahoMessageConverter());

        return adapter;
    }

    /*
        连接工厂
     */
    @Bean
    public MqttPahoClientFactory factory(){
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();

        factory.setConnectionOptions(options());

        return factory;
    }

    /*
        连接工厂的配置信息
     */
    private MqttConnectOptions options() {
        MqttConnectOptions options = new MqttConnectOptions();
        //mqtt服务器地址
        options.setServerURIs(new String[]{"tcp://localhost:1883"});
        //设置登录mqtt服务器用户名
        options.setUserName("qf");
        //设置登录mqtt服务器密码
        options.setPassword("qf".toCharArray());

        //如果这个值为false，那么这个连接会话是一个持久型会话（客户端断开后，会话信息不删除），重连后之前未消费到的消息会继续消费
        //设置为true，每次重连后，之前的信息会丢失
        options.setCleanSession(true);

        return options;
    }
}
